package com.ku.consumerpay.service.mq;

import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.xml.ws.soap.Addressing;
import java.util.concurrent.*;

/**
 * @program: rocketMQKill
 * @description: 事务消息生产者
 * @author: JokerKu
 * @create: 2020-05-06 21:20
 **/
@Component
public class TransactionMQProvider implements InitializingBean {

    @Autowired
    private TransactionListenerImpl transactionListenerImpl;

    private ExecutorService executorService;

    private TransactionMQProducer producer;

    private static final String NAME_SERVER = "192.168.43.10:9876;192.168.43.9:9876";

    private static final String PROVIDER_GROUP_NAME  = "tx_pay_provider_group";

    //私有构造方法初始化MQ参数
    private TransactionMQProvider(){

        this.producer = new TransactionMQProducer(PROVIDER_GROUP_NAME);

        this.executorService = new ThreadPoolExecutor( 2,5,100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1000), new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName(PROVIDER_GROUP_NAME + "-check-thread");
                        return thread;
                    }
                });

        this.producer.setExecutorService(executorService);
        this.producer.setNamesrvAddr(NAME_SERVER);

    }

    //Bean初始化执行的方法---初始化后设置事务监听器
    @Override
    public void afterPropertiesSet() throws Exception {
        this.producer.setTransactionListener(transactionListenerImpl);
        this.producer.start();
    }

    public void shutdown(){
        this.producer.shutdown();
    }

    public TransactionSendResult sendMessage(Message message,Object argument){
        TransactionSendResult sendResult = null;
        try {
            sendResult = this.producer.sendMessageInTransaction(message,argument);
        }catch (Exception e){
            e.printStackTrace();
        }
        return sendResult;
    }
}
